package cn.itcast.job;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockKlineBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.map.SseMap;
import cn.itcast.map.SzseMap;
import cn.itcast.task.*;
import cn.itcast.util.QuotUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;

import java.time.Duration;
import java.util.Properties;

/**
 * 流处理模块:个股业务
 * 子业务包含:
 * 1.秒级行情-hbase
 * 2.分时行情-doris
 * 3.K线行情-doris
 * 4.涨跌幅行情-doris
 * 5.分时数据备份-hdfs
 */
public class StockStream {

    /**
     * 1.创建StockStream对象，创建main方法
     * 2.获取流处理执行环境
     * 3.并行度
     * 4.设置检查点机制
     * 5.设置重启机制
     * 6.整合Kafka(新建反序列化类)
     * 7.数据转换
     * 8.数据过滤（时间和null字段）、合并
     * 9.过滤个股数据
     * 10.设置水位线
     * 11.业务数据处理
     * 12.触发执行
     */
    //1.创建StockStream对象，创建main方法
    public static void main(String[] args) throws Exception {

        /**
         * 数据准备环节（需要对数据提前做处理：数据过滤，清洗脏数据）
         */
        //2.获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //3.并行度
        env.setParallelism(1);
        /**
         * 在开发阶段，当你得代码没有写完成之前，开启检查点备份数据，是没有意义的
         */
        //4.设置检查点机制
        env.enableCheckpointing(5000) ;//每5s制作一次检查的
        env.getCheckpointConfig().setCheckpointStorage("file:///C:\\Users\\Administrator\\Desktop\\code\\checkpoint");//存储路径
        env.getCheckpointConfig().setCheckpointTimeout(60000);//检查点制作超时时间，如果超时，本次制作失败（放弃）
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //精确一次性语义（保证数据只消费一次）
         //当任务取消的时候，保留检查点数据（生产环境使用）
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //5.设置重启策略
        //固定重启2次，每次间隔5s，如果超过2次，程序宕机
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.seconds(5)));

        //6.整合Kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "node1:9092");
        properties.setProperty("group.id", "test"); // 消费者组
        //flink整合kafka
        //消费沪市行情数据，默认是消费最新数据
        FlinkKafkaConsumer<String> kafkaConsumerSse = new FlinkKafkaConsumer<>("sse", new SimpleStringSchema(), properties);
        //消费深市行情数据
        FlinkKafkaConsumer<String> kafkaConsumerSzse = new FlinkKafkaConsumer<>("szse", new SimpleStringSchema(), properties);

        //设置消费模式，从头消费（这种消费模式，只能在开发环境使用，生产环境禁用）
        kafkaConsumerSse.setStartFromEarliest();//如果上生产环境，这两行代码一定要注释掉
        kafkaConsumerSzse.setStartFromEarliest();
        //加载数据源
        DataStreamSource<String> sseSource = env.addSource(kafkaConsumerSse);
        DataStreamSource<String> szseSource = env.addSource(kafkaConsumerSzse);

        //sseSource.print();
//        szseSource.print();

        //7.数据转换
        //sseSource
        SingleOutputStreamOperator<CleanBean> mapSse = sseSource.map(new SseMap());
        //szseSource
        SingleOutputStreamOperator<CleanBean> mapSzse = szseSource.map(new SzseMap());

        //8.数据过滤（时间和null字段）、合并
        //sse
        //在本地测试的时候，注意时间的设置，一定要修改配置文件里面的时间
        SingleOutputStreamOperator<CleanBean> sseFilter = mapSse.filter(new FilterFunction<CleanBean>() {
            @Override
            public boolean filter(CleanBean value) throws Exception {
                return QuotUtil.checkTime(value) && QuotUtil.checkData(value);
            }
        });
        //szse
        SingleOutputStreamOperator<CleanBean> szseFilter = mapSzse.filter(new FilterFunction<CleanBean>() {
            @Override
            public boolean filter(CleanBean value) throws Exception {
                return QuotUtil.checkTime(value) && QuotUtil.checkData(value);
            }
        });

        //数据合并
        DataStream<CleanBean> unionData = sseFilter.union(szseFilter);

        //9.过滤个股数据
        SingleOutputStreamOperator<CleanBean> stockData = unionData.filter(new FilterFunction<CleanBean>() {
            @Override
            public boolean filter(CleanBean value) throws Exception {
                return QuotUtil.isStock(value);
            }
        });

        //10.设置水位线
        SingleOutputStreamOperator<CleanBean> waterData = stockData.assignTimestampsAndWatermarks(WatermarkStrategy.<CleanBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<CleanBean>() {
                    @Override
                    public long extractTimestamp(CleanBean element, long recordTimestamp) {
                        return element.getEventTime();
                    }
                }));
//        waterData.print();

        /**
         * 1.秒级行情-hbase
         * 2.分时行情-doris
         * 3.K线行情-doris
         * 4.涨跌幅行情-doris
         * 5.分时数据备份-hdfs
         */
        //1.秒级行情-hbase
        new StockSecTask().process(waterData);
        //2.分时行情-doris
        new StockMinTask().process(waterData);
        //3.K线行情-doris
        new StockDayKlineTask().process(waterData);
        //4.涨跌幅行情-doris
        new StockRiseAndFallTask().process(waterData);
        //5.分时数据备份-hdfs
        /**
         * 注意事项：
         * 1.写文件到HDFS，需要打开检查点机制
         * 1.1 数据写入过程中有三种状态
         * 1.1.1 正在写入
         * 1.1.2 文件挂起
         * 1.1.3 文件完成
         */
        new StockMinHdfsTask().process(waterData);

        //12.触发执行
        env.execute();
    }

}
